package net.bwie.gd2.dwd;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

public class KafkaJob {
    public static void main(String[] args) {

        //1-表的执行环境
        TableEnvironment tabEnv = getTableEnv();

        //2-输入表-input：映射到Kafka消息队列
        createInputTable(tabEnv);

//
        //3-数据处理-select
        handle(tabEnv);
//
        //4-输出表-output：映射到doris表
        createOutputTable(tabEnv);
//
        //5-将数据保存到doris
        saveToDoris(tabEnv) ;

    }

    private static void saveToDoris(TableEnvironment tabEnv) {

        tabEnv.executeSql(
                "INSERT INTO dws_user_pet_type_score_doris_sink\n" +
                        "select\n" +
                        "    user_id, pet_type,total_score\n" +
                        "from dws_user_pet_type_score"
        );


        tabEnv.executeSql(
                "INSERT INTO dws_user_pet_age_score_doris_sink\n" +
                        "select\n" +
                        "    user_id, pet_type,age_stage,total_score\n" +
                        "from dws_user_pet_type_age_score"
        );


    }

    private static void createOutputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql(
                "CREATE TABLE dws_user_pet_type_score_doris_sink(\n" +
                        "    `user_id` STRING,\n" +
                        "    `pet_type` STRING,\n" +
                        "    `total_score` DOUBLE\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'db_hxl.dws_user_pet_type_score',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        "    'sink.batch.interval' = '10s',\n" +
                        "    'sink.max-retries' = '3',\n" +
                        "    'sink.batch.size' = '1000'\n" +
                        ")"
        );


        tabEnv.executeSql(
                "CREATE TABLE dws_user_pet_age_score_doris_sink(\n" +
                        "    `user_id` STRING,\n" +
                        "    `pet_type` STRING,\n" +
                        "    `age_stage` STRING,\n" +
                        "    `total_score` DOUBLE\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'db_hxl.dws_user_pet_age_score',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        "    'sink.batch.interval' = '10s',\n" +
                        "    'sink.max-retries' = '3',\n" +
                        "    'sink.batch.size' = '1000'\n" +
                        ")"
        );

    }

    private static void handle(TableEnvironment tabEnv) {


        tabEnv.executeSql(
                "CREATE TABLE `dim_pet_category_mapping`(\n" +
                        "    `sub_category` bigint,\n" +
                        "    `pet_type` STRING,\n" +
                        "    `first_category` STRING,\n" +
                        "    `is_general` bigint\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'dim_pet_category',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                        "    'properties.group.id' = 'gid-dim-pet_category',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json',\n" +
                        "    'json.fail-on-missing-field' = 'false',\n" +
                        "    'json.ignore-parse-errors' = 'true'\n" +
                        ")"
        );

//        tabEnv.sqlQuery(
//                "select * from dim_pet_category_mapping"
//        ).execute().print();


        tabEnv.executeSql(
                "CREATE TABLE `dim_pet_age_weight`(\n" +
                        "    `core_category` bigint,\n" +
                        "    `pet_type` STRING,\n" +
                        "    `age_stage` STRING,\n" +
                        "    `weight_buy` double,\n" +
                        "    `weight_add_cart` double,\n" +
                        "    `weight_favorite` double,\n" +
                        "    `weight_pv` double\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'dim_pet_age',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                        "    'properties.group.id' = 'gid-dim-pet_age',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json',\n" +
                        "    'json.fail-on-missing-field' = 'false',\n" +
                        "    'json.ignore-parse-errors' = 'true'\n" +
                        ")"
        );
//
//        tabEnv.sqlQuery(
//                "select * from dim_pet_age_weight"
//        ).execute().print();


        Table dwsUserPetTypeScore = tabEnv.sqlQuery("SELECT\n" +
                "    b.user_id,\n" +
                "    m.pet_type,\n" +
                "    SUM(\n" +
                "                CASE\n" +
                "                    WHEN behavior_type='buy' THEN 5\n" +
                "                    WHEN behavior_type='add_cart' THEN 3\n" +
                "                    WHEN behavior_type='favorite' THEN 2\n" +
                "                    WHEN behavior_type='pv' THEN 1\n" +
                "                    END *\n" +
                "                CASE WHEN m.is_general=1 THEN 0.5 ELSE 1 END\n" +
                "        ) AS total_score\n" +
                "FROM dwd_user_product_log b\n" +
                "         JOIN dim_pet_category_mapping m\n" +
                "              ON b.category_id = m.sub_category\n" +
                "WHERE b.behavior_time >= TIMESTAMPADD(DAY, -180, CURRENT_TIMESTAMP)\n" +
                "GROUP BY b.user_id, m.pet_type");

        tabEnv.createTemporaryView("dws_user_pet_type_score", dwsUserPetTypeScore);

//        tabEnv.sqlQuery(
//                "select * from dws_user_pet_type_score"
//        ).execute().print();

        tabEnv.executeSql(
                "CREATE TEMPORARY VIEW base_data AS\n" +
                        "SELECT\n" +
                        "    b.user_id,\n" +
                        "    b.product_id,\n" +
                        "    a.pet_type,\n" +
                        "    a.age_stage,\n" +
                        "    CASE\n" +
                        "        WHEN b.behavior_type = 'buy' THEN a.weight_buy\n" +
                        "        WHEN b.behavior_type = 'add_cart' THEN a.weight_add_cart\n" +
                        "        WHEN b.behavior_type = 'favorite' THEN a.weight_favorite\n" +
                        "        ELSE a.weight_pv\n" +
                        "        END AS base_weight,\n" +
                        "    TIMESTAMPDIFF(DAY, TO_TIMESTAMP(b.behavior_time, 'yyyy-MM-dd HH:mm:ss'), CAST(CURRENT_TIMESTAMP AS TIMESTAMP)) AS diff_days,\n" +
                        "    TO_TIMESTAMP(b.behavior_time, 'yyyy-MM-dd HH:mm:ss') AS behavior_time -- 保留时间字段用于排序\n" +
                        "FROM dwd_user_product_log b\n" +
                        "         JOIN dim_pet_age_weight a ON \n" +
                        "            b.category_id = a.core_category \n" +
                        "WHERE\n" +
                        "         b.behavior_time >= TIMESTAMPADD(DAY, -180, CURRENT_TIMESTAMP)\n" +
                        "    AND a.pet_type IN ('狗', '猫')"
        );



        tabEnv.executeSql(
                "CREATE TEMPORARY VIEW with_decay AS\n" +
                        "SELECT\n" +
                        "    user_id,\n" +
                        "    product_id,\n" +
                        "    pet_type,\n" +
                        "    age_stage,\n" +
                        "    base_weight *\n" +
                        "    POWER(0.7, 1) *\n" +
                        "    POWER(0.9, FLOOR(diff_days / 30)) AS decayed_weight\n" +
                        "FROM base_data"
        );

//                tabEnv.sqlQuery(
//                "select * from with_decay"
//        ).execute().print();
//
//
        Table table = tabEnv.sqlQuery(
                "SELECT\n" +
                        "    user_id,\n" +
                        "    pet_type,\n" +
                        "    age_stage,\n" +
                        "    SUM(decayed_weight) AS total_score\n" +
                        "FROM with_decay\n" +
                        "GROUP BY user_id, pet_type, age_stage"
        );
        tabEnv.createTemporaryView("dws_user_pet_type_age_score", table);
//        tabEnv.sqlQuery(
//                "select * from dws_user_pet_type_age_score"
//        ).execute().print();



    }


    //todo 映射到Kafka消息队列
    private static void createInputTable(TableEnvironment tabEnv) {

        tabEnv.executeSql(
                "CREATE TABLE `dwd_user_product_log`(\n" +
                        "    `user_id` STRING,\n" +
                        "    `product_id` STRING,\n" +
                        "    `behavior_type` STRING,\n" +
                        "    `behavior_time` string,\n" +
                        "    `category_id` bigint,\n" +
                        "    `category_name` STRING,\n" +
                        "    `first_category` STRING,\n" +
                        "    `title` STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'dwd_pet_behavior',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                        "    'properties.group.id' = 'gid-dwd-user_product_log',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json',\n" +
                        "    'json.fail-on-missing-field' = 'false',\n" +
                        "    'json.ignore-parse-errors' = 'true'\n" +
                        ")"
        );

//        tabEnv.sqlQuery(
//                "select * from dwd_user_product_log"
//        ).execute().print();

    }


    //todo 1-表的执行环境
    private static TableEnvironment getTableEnv() {
        //1-环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        //2-配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");

        //3-返回对象
        return tabEnv;
    }

}
